package com.ctyun.xstore.vod.video;

import static com.ctyun.xstore.vod.video.model.VideoFileInfoBean.DEFAULT_PART_SIZE;
import static com.ctyun.xstore.vod.video.model.VideoFileInfoBean.PART_SIZE_MAX;
import static com.ctyun.xstore.vod.video.model.VideoFileInfoBean.PART_SIZE_MIN;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ctyun.xstore.vod.ConcurrentPartUploader;
import com.ctyun.xstore.vod.video.model.PartETag;

public class ContinuoursVideoUploader {

	private static final int DEFAULT_MAX_POOL_SIZE = 4;

	private final File file;

	private final List<String> urls = new LinkedList<>();

	private final int partSize;

	private final ConcurrentPartUploader s3urlUploader;

	private final File infoTempfile;

	private final Map<String, PartETag> partETags = new HashMap<>();

	private final ExecutorService executeService;

	private ContinuoursVideoUploader(File file, List<String> urls, int partSize, int maxAsyncThreadsCount,
			String tmpDir) {
		super();
		this.file = file;
		if (urls != null)
			this.urls.addAll(urls);

		if (partSize < PART_SIZE_MIN || partSize > PART_SIZE_MAX) {
			throw new IllegalArgumentException("Part size should between 5M to 100M.");
		}

		this.partSize = partSize;

		if (maxAsyncThreadsCount <= 0)
			throw new IllegalArgumentException("");

		executeService = new ThreadPoolExecutor(0, maxAsyncThreadsCount, 0, TimeUnit.MILLISECONDS,
				new LinkedBlockingQueue<Runnable>());

		this.s3urlUploader = new ConcurrentPartUploader();

		String tDir = tmpDir != null ? tmpDir : System.getProperty("java.io.tmpdir");

		infoTempfile = new File(tDir, String.format("uploading_%d", file.getAbsolutePath().hashCode()));
	}

	private void loadSavedInfos() throws IOException {
		if (!infoTempfile.exists() || !infoTempfile.isFile()) {
			initTempFile();
			return;
		}
		try (BufferedReader reader = new BufferedReader(new FileReader(infoTempfile))) {
			// fist line
			String line = reader.readLine();
			if (line == null || !this.isFileInfoMatch(line)) {
				initTempFile();
				return;
			}
			while ((line = reader.readLine()) != null) {
				this.addTempPartETag(line);
			}
		}
	}

	private void addTempPartETag(String jsonStr) {
		if (jsonStr.isEmpty())
			return;
		try {
			PartETag partETag = JSON.toJavaObject(JSON.parseObject(jsonStr), PartETag.class);
			this.partETags.put(String.valueOf(partETag.getPartNumber()), partETag);
		} catch (Exception e) {
			// do nothing
		}
	}

	private boolean isFileInfoMatch(String jsonStr) {
		try {
			JSONObject json = JSON.parseObject(jsonStr);
			long fileSize = json.getLongValue("fileSize");
			if (fileSize != file.length())
				return false;
			JSONArray list = json.getJSONArray("urls");
			if (this.urls.isEmpty()) {
				this.urls.addAll(list.toJavaList(String.class));
				return true;
			}
			if (list.size() != this.urls.size())
				return false;
			for (int i = 0; i < list.size(); i++) {
				if (!list.getString(i).equals(this.urls.get(i))) {
					return false;
				}
			}

		} catch (Exception e) {
			return false;
		}
		return true;
	}

	private void initTempFile() throws IOException {
		Files.deleteIfExists(Paths.get(infoTempfile.getAbsolutePath()));
		infoTempfile.createNewFile();
		this.writeFileInfoToTempFile();
	}

	private void writeFileInfoToTempFile() throws IOException {
		try (FileOutputStream out = new FileOutputStream(infoTempfile)) {
			JSONObject fileInfo = new JSONObject();
			fileInfo.put("fileSize", file.length());
			fileInfo.put("urls", this.urls);
			out.write(fileInfo.toJSONString().getBytes());
		}
	}

	public List<PartETag> upload() throws IOException {
		this.loadSavedInfos();
		this.checkUrls();
		return this.doUpload();
	}

	private void checkUrls() throws IOException {
		if (this.urls.isEmpty())
			throw new IOException("Can load urls from saved files, it seems that you have not upload this file["
					+ this.file.getName() + "] before.");
	}

	private List<PartETag> doUpload() throws IOException {
		try (FileInputStream fin = new FileInputStream(file);
				FileChannel channel = fin.getChannel();
				BufferedWriter out = new BufferedWriter(new FileWriter(infoTempfile, true))) {
			ByteBuffer buffer = ByteBuffer.allocate(partSize);
			int index = 0;
			int len = 0;

			List<Future<String>> results = new LinkedList<>();

			while ((len = (channel.read(buffer))) != -1) {

				final byte[] array = new byte[len];
				((Buffer) buffer).flip();
				buffer.get(array);
				((Buffer) buffer).clear();

				String url = urls.get(index);
				index++;
				final int partNum = index;

				if (this.partETags.containsKey(String.valueOf(partNum))) {
					continue;
				}

				Future<String> res = this.executeService.submit(() -> {
					try {
						String eTag = s3urlUploader.upload(url, array);
						PartETag partETag = new PartETag(partNum, eTag);
						partETags.put(String.valueOf(partNum), partETag);
						synchronized (out) {
							out.newLine();
							out.write(JSON.toJSONString(partETag));
						}
						return "";
					} catch (Exception e) {
						return "Upload part[" + partNum + "] fails, error message: " + e.getMessage() + "\n";
					}
				});
				results.add(res);
			}

			this.executeService.shutdown();
			String errorMsg = "";
			for (Future<String> res : results) {
				try {
					errorMsg += res.get();
				} catch (InterruptedException | ExecutionException e) {
					errorMsg += e.getMessage();
				}
			}
			if (!"".equals(errorMsg))
				throw new IOException(errorMsg);
		}
		Files.delete(this.infoTempfile.toPath());
		return partETags.values().stream().collect(Collectors.toList());
	}

	public static final class Builder {

		private File file;

		private List<String> urls;

		private int partSize = DEFAULT_PART_SIZE;

		private int maxAsyncThreadsCount = DEFAULT_MAX_POOL_SIZE;

		private String tmpDir;

		public Builder() {
			super();
		}

		public Builder(String filePath) {
			this(filePath, Collections.emptyList());
		}

		public Builder(File file) {
			this(file, Collections.emptyList());
		}

		public Builder(String filePath, List<String> urls) {
			this(new File(filePath), urls);
		}

		public Builder(File file, List<String> urls) {
			this.setFile(file).setUrls(urls);
		}

		public Builder setFilePath(String path) {
			this.file = new File(path);
			return this;
		}

		public Builder setFile(File file) {
			this.file = file;
			return this;
		}

		public Builder setUrls(List<String> urls) {
			this.urls = urls;
			return this;
		}

		public Builder setPartSize(int partSize) {
			this.partSize = partSize;
			return this;
		}

		public Builder setMaxAsyncThreadsCount(int maxAsyncThreadsCount) {
			this.maxAsyncThreadsCount = maxAsyncThreadsCount;
			return this;
		}

		public Builder setTmpDir(String tmpDir) {
			this.tmpDir = tmpDir;
			return this;
		}

		public ContinuoursVideoUploader build() {
			return new ContinuoursVideoUploader(file, urls, partSize, maxAsyncThreadsCount, tmpDir);
		}

	}
}
